package org.ala.tiktools.rpc.kcp.netty.server.handler;

import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.concurrent.TimeUnit;

import org.ala.tiktools.logs.LoggerFactory;
import org.ala.tiktools.rpc.kcp.exception.KcpRpcException;
import org.ala.tiktools.rpc.kcp.netty.KcpPacket;
import org.ala.tiktools.rpc.kcp.netty.NettyKcp;
import org.ala.tiktools.rpc.kcp.netty.server.KcpServerChannel;
import org.ala.tiktools.rpc.kcp.netty.server.KcpServerChildChannel;
import org.ala.tiktools.rpc.kcp.netty.tools.NettyTools;
import org.ala.tiktools.tools.TimestampTools;
import org.slf4j.Logger;

import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelOutboundBuffer;
import io.netty.channel.socket.DatagramChannel;


/**
 * ServerChannel负责的Writer
 *
 * @author ala
 * @date 2025-04-15 16:59
 */
public class KcpServerChannelWriter {
	static Logger log = LoggerFactory.getLogger("rpc", "server");
	
	
    /**
     *  上次update时间戳
     */
    protected int tsUpdate;
    /**
     *  update正在执行中标记位
     */
    protected boolean scheduleUpdate;
    /**
     *  定时任务体
     */
    protected ScheduleUpdate scheduleUpdater;


    /**
     *
     */
    protected KcpServerChannel serverChannel;
    public KcpServerChannelWriter(KcpServerChannel serverChannel) {
        this.serverChannel = serverChannel;
        this.scheduleUpdater = new ScheduleUpdate(serverChannel, this);
    }


    /**
     *  通过javaChannel吧数据真实发送出去
     */
    public boolean doWriteMessage(Object msg, ChannelOutboundBuffer in) throws Exception {
        KcpPacket packet = (KcpPacket) msg;
        InetSocketAddress addr = packet.getRemoteAddress();
        ByteBuf data = packet.getContent();

        int len = data.readableBytes();
        if (len == 0) { return true; }

        ByteBuffer butter = data.internalNioBuffer(data.readerIndex(), len);
        //  调用的是KcpServerChildChannel.send()
//        log.info("[KcpServerChannelWriter doWriteMessage] 服务端发送消息 len:{} byte  addr:{}", len, addr);
        return serverChannel.javaChannel().send(butter, addr) > 0;
    }


    /**
     *  update子channel
     */
    public void updateChildKcp(KcpServerChildChannel childChannel) {
        int current = TimestampTools.milliSeconds32();
        Throwable ex = null;
        try {
            childChannel.getNettyKcp().update(current);
        } catch (Exception e) { ex = e; }
        if (childChannel.getNettyKcp().getState() == -1 && ex == null) {
            ex = new KcpRpcException("state = -1 after update. kcp:" + childChannel.getNettyKcp());
        }
        if (ex != null) {
            NettyTools.fireExceptionAndClose(childChannel, ex);
        }
    }


    /**
     *  定时触发update
     *  <p> 会触发serverChannel管理的所有子channel的update
     */
    public void scheduleUpdate(int tsUpdate, int current) {
        this.tsUpdate = tsUpdate;
        this.scheduleUpdate = true;
        //  定时跑flush（eventLoop是boss线程）
        serverChannel.eventLoop().schedule(scheduleUpdater, tsUpdate - current, TimeUnit.MILLISECONDS);
    }
    /**
     *  update任务体
     */
    public static class ScheduleUpdate implements Runnable {
        /**
         *  触发update过程中，发现要出发flush的channel
         */
        protected List<KcpServerChildChannel> writeChildChannelList = new ArrayList<>();
        /**
         *  过程中需要关闭的list
         */
        protected List<KcpServerChildChannel> closedChildChannelList = new ArrayList<>();
        protected KcpServerChannel serverChannel;
        protected KcpServerChannelWriter writer;
        public ScheduleUpdate(KcpServerChannel serverChannel, KcpServerChannelWriter writer) {
            this.serverChannel = serverChannel;
            this.writer = writer;
        }
        @Override
        public void run() {
            int current = TimestampTools.milliSeconds32();
            int nextTsUpdate = 0;
            boolean nextSchedule = false;
            //  遍历所有的子channel
            for (KcpServerChildChannel channel : serverChannel.getChildChannelMap().values()) {
                if (!channel.isActive()) { continue; }

                NettyKcp nkcp = channel.getNettyKcp();
                int tsUpdate = nkcp.getTsUpdate();
                int nextChildTsUp = 0;
                boolean nextChildSchedule = false;
                //  如果已经过了触发时间，直接触发
                if (current >= tsUpdate) {
                    Throwable ex = null;
                    try {
                        nkcp.update(current);
                        nextChildSchedule = true;
                    } catch (Throwable e) {
                        ex = e;
                    }
                    //  如果执行完以后状态变-1了，判定执行失败
                    if (nkcp.getState() == -1 && ex == null) {
                        ex = new KcpRpcException(String.format("state == -1 after update. kcp:%s", nkcp));
                    }

                    if (ex != null) {
                        closedChildChannelList.add(channel);
                        nextChildTsUp = 0;
                        nextChildSchedule = false;
                    }
                    //  如果update执行成功，尝试执行一次flush
                    else if (channel.isFlushPending() && channel.kcpCanSend()) {
                        writeChildChannelList.add(channel);
                    }
                }
                //  否则记录下次触发时间
                else {
                    nextChildTsUp = tsUpdate;
                    nextChildSchedule = true;
                }
                //  记录下次触发update的时间戳（本轮触发update的子channel的最大下次触发时间戳）
                if (nextChildSchedule && (!nextSchedule || nextChildTsUp > nextTsUpdate)) {
                    nextSchedule = true;
                    nextTsUpdate = nextChildTsUp;
                }
            }
            //  尝试执行flush
            writeChildChannelList.forEach(c -> {
                c.unsafe().forceFlush();
            });
            writeChildChannelList.clear();

            //  如果存在等待关闭的channel，关闭之前触发他们的update，把消息发送出去
            if (!serverChannel.getCloser().getCloseWaitMap().isEmpty()) {
                List<InetSocketAddress> removes = new ArrayList<>();
                for (Map.Entry<InetSocketAddress, KcpServerChannelCloser.CloseWaitTask> e : serverChannel.getCloser().getCloseWaitMap().entrySet()) {
                    InetSocketAddress k = e.getKey();
                    NettyKcp nkcp = e.getValue().getNkcp();
                    int tsUp = nkcp.getTsUpdate();
                    int nextChildTsUp = 0;
                    boolean nextChildSchedule = false;
                    Throwable exception = null;
                    if (current >= tsUp) {
                        try {
                            nextChildTsUp = nkcp.update(current);
                            nextChildSchedule = true;
                        } catch (Throwable t) {
                            exception = t;
                            removes.add(k);
                            nkcp.setKcpClosed();
                            KcpServerChannel.log.error("Terminate closeWaitKcp. ukcp={}, cause={}", nkcp, "update error", t);
                        }
                        if (nkcp.getState() == -1 && exception == null) {
                            removes.add(k);
                            nkcp.setKcpClosed();
                            nextChildTsUp = 0;
                            nextChildSchedule = false;
                            if (KcpServerChannel.log.isDebugEnabled()) {
                            	KcpServerChannel.log.debug("Terminate closeWaitKcp. ukcp={}, cause={}", nkcp, "update -1");
                            }
                        }
                    } else {
                        nextChildTsUp = tsUp;
                        nextChildSchedule = true;
                    }
                    if (nextChildSchedule && (!nextSchedule || nextTsUpdate > nextChildTsUp)) {
                        nextTsUpdate = nextChildTsUp;
                        nextSchedule = true;
                    }
                }
            }

            //  设置下次flush时间戳
            writer.tsUpdate = nextTsUpdate;
            writer.scheduleUpdate = nextSchedule;
            if (nextSchedule) { writer.scheduleUpdate(nextTsUpdate, current); }

            //  过程中如果有需要close的
            serverChannel.getCloser().handleCloseChildChannelList(closedChildChannelList);
        }
    }

    public boolean isScheduleUpdate() {
        return scheduleUpdate;
    }

    public void setScheduleUpdate(boolean scheduleUpdate) {
        this.scheduleUpdate = scheduleUpdate;
    }
}
